
package com.zliuyno.io.testreactor;

import com.bjmashibing.system.rpcdemo.rpc.protocol.Myheader;
import com.sun.org.apache.regexp.internal.RE;
import com.sun.xml.internal.bind.v2.runtime.reflect.Lister;
import io.netty.bootstrap.Bootstrap;
import io.netty.bootstrap.ServerBootstrap;
import io.netty.buffer.ByteBuf;
import io.netty.buffer.PooledByteBufAllocator;
import io.netty.channel.*;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.nio.NioServerSocketChannel;
import io.netty.channel.socket.nio.NioSocketChannel;
import io.netty.handler.codec.ByteToMessageDecoder;
import org.junit.Test;

import javax.swing.text.Keymap;
import java.io.*;
import java.lang.reflect.InvocationHandler;
import java.lang.reflect.Method;
import java.lang.reflect.Proxy;
import java.net.InetSocketAddress;
import java.rmi.UnexpectedException;
import java.util.List;
import java.util.Map;
import java.util.Random;
import java.util.UUID;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.atomic.AtomicInteger;

/**
 * @author liuyno
 * @date 2020/12/2 - 17:56
 * <p>
 * 1.写一个RPC：就像调用本地方法一样调用远程方法，面向java中就是所谓的 面向interface开发
 * 2.来回通信，连接数量  拆包 粘包
 * 3.动态代理 序列化  协议封装
 * 4.连接池
 */

public class MyRPCTest {
    @Test
    public void get() {

        AtomicInteger num = new AtomicInteger(0);

        new Thread(() -> {
            startServer();
        }).start();

        System.out.println("server started...");

        int size = 20;
        Thread[] threads = new Thread[size];
        for (int i = 0; i < size; i++) {
            threads[i] = new Thread(() -> {
                Car car = proxyGet(Car.class);   //动态代里实现
                car.ooxx("Hello" + num.incrementAndGet());
            });
        }

        for (Thread thread : threads) {
            thread.start();
        }

        try {
            System.in.read();
        } catch (IOException e) {
            e.printStackTrace();
        }

//        Fly fly = proxyGet(Fly.class);   //动态代里实现
//        fly.xxoo("Hello");
    }

    @Test
    public void startServer() {
        NioEventLoopGroup group = new NioEventLoopGroup(1);
        ServerBootstrap bs = new ServerBootstrap();
        ChannelFuture bind = bs.group(group, group)
                .channel(NioServerSocketChannel.class)
                .childHandler(new ChannelInitializer<NioSocketChannel>() {
                    @Override
                    protected void initChannel(NioSocketChannel ch) throws Exception {
                        System.out.println("server accept client port :" + ch.remoteAddress().getPort());
                        ChannelPipeline pipeline = ch.pipeline();
                        pipeline.addLast(new ServerDecode());
                        pipeline.addLast(new MyServerHandler());
                    }
                }).bind(new InetSocketAddress("localhost", 9090));
        try {
            bind.sync().channel().closeFuture().sync();

        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    }


    //模拟counsumer端
    public static <T> T proxyGet(Class<T> interfaceInfo) {
        //实现各个版本的动态代理

        ClassLoader loader = interfaceInfo.getClassLoader();
        Class<?>[] methodInfo = {interfaceInfo};


        return (T) Proxy.newProxyInstance(loader, methodInfo, new InvocationHandler() {
            @Override
            public Object invoke(Object proxy, Method method, Object[] args) throws Throwable {
                //如何设计我们的consumer 对于 provider的调用过程

                //1.调用服务 、方法、参数 =》 封装成 message [content]
                String name = interfaceInfo.getName();
                String methodName = method.getName();
                Class<?>[] parameterTypes = method.getParameterTypes();

                MyContent content = new MyContent();
                content.setArgs(args);
                content.setName(name);
                content.setMethodName(methodName);
                content.setParameterTypes(parameterTypes);

                ByteArrayOutputStream out = new ByteArrayOutputStream();
                ObjectOutputStream oout = new ObjectOutputStream(out);
                oout.writeObject(content);
                byte[] msgBody = out.toByteArray();

                //2.requestID+message   本地要缓存 requestID
                //协议：[header<>] [msgBody]
                MyHeader header = createHeader(msgBody);
                out.reset();
                oout = new ObjectOutputStream(out);
                oout.writeObject(header);
                byte[] msgHeader = out.toByteArray();
//                System.out.println("msgHeader :" + msgHeader.length);

                //3.连接池 ：取得连接
                ClientFactory factory = ClientFactory.getFactory();
                NioSocketChannel clientChannel = factory.getClient(new InetSocketAddress("localhost", 9090));

                //获取连接的过程   开始-创建   连接中-直接取
                //4.发送--> 走 IO out --> 走Netty（event 驱动）
                ByteBuf byteBuf = PooledByteBufAllocator.DEFAULT.directBuffer(msgHeader.length + msgBody.length);
                CountDownLatch countDownLatch = new CountDownLatch(1);
                long requestID = header.getRequestID();
                ResponseHandler.addCallBack(requestID, new Runnable() {
                    @Override
                    public void run() {
                        countDownLatch.countDown();
                    }
                });
                byteBuf.writeBytes(msgHeader);
                byteBuf.writeBytes(msgBody);
                ChannelFuture channelFuture = clientChannel.writeAndFlush(byteBuf);
                channelFuture.sync();   //IO是双向的 你看似有个sync 她仅代表out


                countDownLatch.await();


                //5.如果从IO 未来回来了 怎么将代码执行到这里
                //（睡眠/回调，如何让线程停下来，还能让他继续）

                return null;
            }
        });
    }

    private static MyHeader createHeader(byte[] msg) {
        MyHeader header = new MyHeader();
        int size = msg.length;
        int f = 0x14141414;
        long requestID = Math.abs(UUID.randomUUID().getLeastSignificantBits());
        header.setFlag(f);
        header.setDataLen(size);
        header.setRequestID(requestID);
        return header;
    }
}


class ServerDecode extends ByteToMessageDecoder {
    //父类里一定有channelRead被调起，
    @Override
    protected void decode(ChannelHandlerContext channelHandlerContext, ByteBuf buf, List<Object> list) throws Exception {
//        System.out.println("channel start:" + buf.readableBytes());
        while (buf.readableBytes() >= 105) {
            byte[] bytes = new byte[105];
            buf.getBytes(buf.readerIndex(), bytes);  //读取内容时指针不发生移动
            ByteArrayInputStream in = new ByteArrayInputStream(bytes);
            ObjectInputStream oin = new ObjectInputStream(in);
            MyHeader header = (MyHeader) oin.readObject();
//            System.out.println("server response @ id:" + header.getRequestID());

            //server端代码  取body
            if (buf.readableBytes() >= header.getDataLen()) {
                buf.readBytes(105); //移动指针到body开始的位置
                byte[] data = new byte[(int) header.getDataLen()];
                buf.readBytes(data);
                ByteArrayInputStream din = new ByteArrayInputStream(data);
                ObjectInputStream doin = new ObjectInputStream(din);
                MyContent content = (MyContent) doin.readObject();
//                System.out.println(content.getName());
                list.add(new Packmsg(header, content));
            } else {
                break;
            }
        }
    }
}

class Packmsg {
    MyHeader header;
    MyContent content;

    public Packmsg(MyHeader header, MyContent content) {
        this.header = header;
        this.content = content;
    }

    public MyHeader getHeader() {
        return header;
    }

    public void setHeader(MyHeader header) {
        this.header = header;
    }

    public MyContent getContent() {
        return content;
    }

    public void setContent(MyContent content) {
        this.content = content;
    }
}

class MyServerHandler extends ChannelInboundHandlerAdapter {
    @Override
    public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
        Packmsg requestPkg = (Packmsg) msg;
        System.out.println("server handler:" + requestPkg.content.getArgs()[0]);

        //如果处理完了 需要给客户端返回
        //返回数据需要封装为一个bytebuf    因为是RPC,你得有requestID
        //在client一侧需要解决编码问题

        //有新的header+content
        String ioThreadName = Thread.currentThread().getName();
        //1.直接在当前方法处理IO 业务和处理返回的事
        //2.使用netty自己的eventloop来处理业务及返回
        ctx.executor().execute(new Runnable() {
            @Override
            public void run() {
                String execThreadName = Thread.currentThread().getName();

                MyContent content = new MyContent();
                String s = "io thread:" + ioThreadName + "exec thread:" + execThreadName + "from args" + requestPkg.content.getArgs()[0];
                content.setRes(s);


                MyHeader header = new MyHeader();
                header.setRequestID(requestPkg.getHeader().getRequestID());
                header.setFlag(0x14141424);
                header.setDataLen(s.length());
            }
        });
    }
}


class ResponseHandler {

    static ConcurrentHashMap<Long, Runnable> mapping = new ConcurrentHashMap<>();

    public static void addCallBack(long requestID, Runnable cb) {
        mapping.putIfAbsent(requestID, cb);
    }

    public static void runCallBack(long requestID) {

        Runnable runnable = mapping.get(requestID);

        runnable.run();
        removeCB(requestID);
    }

    private static void removeCB(long requestID) {
        mapping.remove(requestID);
    }

}

class ClientFactory {

    int poolSize = 1;

    NioEventLoopGroup group;

    Random random = new Random();

    private ClientFactory() {
    }

    private static final ClientFactory factory;

    static {
        factory = new ClientFactory();
    }

    public static ClientFactory getFactory() {
        return factory;
    }

    //一个consumer可以连接多个provider,每一个provider都有自己的pool   K,V
    ConcurrentHashMap<InetSocketAddress, ClientPool> outboxs = new ConcurrentHashMap<>();


    public NioSocketChannel getClient(InetSocketAddress address) {
        ClientPool clientPool = outboxs.get(address);
        if (clientPool == null) {
            outboxs.putIfAbsent(address, new ClientPool(poolSize));
            clientPool = outboxs.get(address);
        }
        int i = random.nextInt(poolSize);
        if (clientPool.clients[i] != null && clientPool.clients[i].isActive()) {
            return clientPool.clients[i];
        }

        synchronized (clientPool.lock[i]) {
            return clientPool.clients[i] = create(address);
        }

    }

    private NioSocketChannel create(InetSocketAddress address) {
        //基于netty的客户端创建方式
        group = new NioEventLoopGroup(1);
        Bootstrap bs = new Bootstrap();
        ChannelFuture connect = bs.group(group)
                .channel(NioSocketChannel.class)
                .handler(new ChannelInitializer<NioSocketChannel>() {
                    @Override
                    protected void initChannel(NioSocketChannel nioSocketChannel) throws Exception {
                        ChannelPipeline p = nioSocketChannel.pipeline();
                        p.addLast(new ClientResonses());    //解决给谁的 谁去处理这个返回
                    }
                }).connect(address);
        Channel client = null;
        try {
            client = connect.sync().channel();
            return (NioSocketChannel) client;
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
        return null;
    }
}


class ClientResonses extends ChannelInboundHandlerAdapter {

    @Override
    public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
        ByteBuf buf = (ByteBuf) msg;
        if (buf.readableBytes() >= 105) {
            byte[] bytes = new byte[105];
            buf.readBytes(bytes);
            ByteArrayInputStream in = new ByteArrayInputStream(bytes);
            ObjectInputStream oin = new ObjectInputStream(in);
            MyHeader header = (MyHeader) oin.readObject();
            System.out.println("client response @ id" + header.getRequestID());

            //TODO:
            ResponseHandler.runCallBack(header.requestID);

            //server端代码  取body
            /*if (buf.readableBytes() >= header.getDataLen()) {
                byte[] data = new byte[(int) header.getDataLen()];
                buf.readBytes(data);
                ByteArrayInputStream din = new ByteArrayInputStream(data);
                ObjectInputStream doin = new ObjectInputStream(din);
                MyContent content = (MyContent) doin.readObject();
                System.out.println(content.getName());
            }*/

        }


    }
}


class ClientPool {
    NioSocketChannel[] clients;
    Object[] lock;

    ClientPool(int size) {
        clients = new NioSocketChannel[size];      //inti 连接是空的
        lock = new Object[size];    //锁是可以初始化的
        for (int i = 0; i < size; i++) {
            lock[i] = new Object();
        }

    }
}


class MyHeader implements Serializable {
    //通信上的协议
    /**
     * 1.ooxx值
     * 2.UUID
     * 3.DATA_LEN
     */
    int flag;   //32bit可以设置很多信息
    long requestID;
    long dataLen;

    public int getFlag() {
        return flag;
    }

    public void setFlag(int flag) {
        this.flag = flag;
    }

    public long getRequestID() {
        return requestID;
    }

    public void setRequestID(long requestID) {
        this.requestID = requestID;
    }

    public long getDataLen() {
        return dataLen;
    }

    public void setDataLen(long dataLen) {
        this.dataLen = dataLen;
    }
}

class MyContent implements Serializable {
    String name;
    String methodName;
    Class<?>[] parameterTypes;
    Object[] args;
    String res;


    public String getRes() {
        return res;
    }

    public void setRes(String res) {
        this.res = res;
    }

    public String getName() {
        return name;
    }

    public void setName(String name) {
        this.name = name;
    }

    public String getMethodName() {
        return methodName;
    }

    public void setMethodName(String methodName) {
        this.methodName = methodName;
    }

    public Class<?>[] getParameterTypes() {
        return parameterTypes;
    }

    public void setParameterTypes(Class<?>[] parameterTypes) {
        this.parameterTypes = parameterTypes;
    }

    public Object[] getArgs() {
        return args;
    }

    public void setArgs(Object[] args) {
        this.args = args;
    }
}


interface Car {

    public void ooxx(String msg);

}

interface Fly {

    public void xxoo(String msg);
}
